package com.voice.client.message.impl;

import com.voice.client.AbstractClientImpl;
import com.voice.client.channel.tcp.TCPServer;
import com.voice.client.message.MessageClient;
import com.voice.client.message.entity.SubscribeInfo;
import com.voice.client.message.callback.MessageCallback;
import com.voice.client.message.entity.Message;
import com.voice.client.Profile;
import com.voice.core.entity.HttpHeader;
import com.voice.core.connection.Connection;
import com.voice.core.connection.entity.Stream;
import com.voice.core.connection.entity.StreamData;
import com.voice.core.exception.ClientException;
import com.voice.core.utils.MessageUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class MessageClientImpl extends AbstractClientImpl implements MessageClient{

    private MessageCallback messageCallback;
    private AtomicLong localMessageId;
    private List<SubscribeInfo> subscribeInfos;

    public MessageClientImpl(Profile profile) {
        super(profile);
        localMessageId=new AtomicLong(0);
        this.subscribeInfos=new ArrayList<>();
    }

    @Override
    public String path() {
        return "/message/echo/success";
    }



    @Override
    public void connect(MessageCallback messageCallback) {
        if (this.isConnected.compareAndSet(false, true)) {
            this.setMessageListener(messageCallback);
            doConnect();
        } else {
            throw new ClientException("client is already connected!");
        }
    }


    @Override
    public void setMessageListener(MessageCallback messageCallback) {
        this.messageCallback = messageCallback;
    }

    @Override
    public void setMessageListener(String topic, MessageCallback messageCallback) {
        this.messageCallback = messageCallback;
        Optional<SubscribeInfo> optional = this.subscribeInfos.stream().filter((info) -> info.isMatch(topic)).findAny();
        if (optional.isPresent()) {
            if (messageCallback == null) {
                this.subscribeInfos.remove(optional.get());
            } else {
                optional.get().setCallback(messageCallback);
            }
        } else {
            this.subscribeInfos.add(new SubscribeInfo(topic, messageCallback));
        }
    }

    private CompletableFuture<Boolean> sendWithResult(HttpHeader headers, byte[] bytes, String errorMessage){
        CompletableFuture<Boolean> future=new CompletableFuture<>();
        this.client().sendRequest(connection(),headers,bytes).whenComplete((response,throwable)->{
            if (throwable != null) {
                log.error(errorMessage, throwable);
                future.completeExceptionally(throwable);
            } else {
                if (!response.isOk()) {
                    ClientException exception = new ClientException(errorMessage, response);
                    log.error(exception.getMessage());
                    future.completeExceptionally(exception);
                } else {
                    future.complete(true);
                }
            }
        });
        return future;
    }
    @Override
    public CompletableFuture<Boolean> subscribe(String topic) {
        this.checkStarted();
        HttpHeader headers = new HttpHeader();
        headers.path("/message/sub/" + topic);
        return this.sendWithResult(headers, null, "failed to subscribe " + topic);
    }

    @Override
    public CompletableFuture<Boolean> subscribe(String topic, MessageCallback callback) {
        this.setMessageListener(topic, callback);
        return this.subscribe(topic);
    }

    @Override
    public CompletableFuture<Boolean> unsubscribe(String topic) {
        this.checkStarted();
        HttpHeader headers = new HttpHeader();
        headers.path("/message/unsub/" + topic);
        return this.sendWithResult(headers, null, "failed to unsubscribe " + topic);
    }

    @Override
    public CompletableFuture<Message> publish(String topic, Message message) {
        this.checkStarted();
        HttpHeader headers = new HttpHeader();
        headers.path("/message/pub" + topic);
        headers.put("x-qos", message.getQos());
        headers.put("x-generate-time", message.getGenerateTime());
        headers.put("x-message-id", this.localMessageId.getAndIncrement());
        CompletableFuture<StreamData> completableFuture=this.client().sendRequest(connection(), headers, message.getPayload());
        CompletableFuture<Message> cf=new CompletableFuture<>();
        completableFuture.whenComplete(((streamData, throwable) -> {
            if(throwable!=null) cf.completeExceptionally(throwable);
            else {
                Message m=convertStreamData2Message(streamData);
                cf.complete(m);
            }
        }));
        return cf;
    }

    @Override
    public void publishServerPort(int port) throws IOException {
        this.server=new TCPServer(port,this);
        this.server.start();
        String topic="/server/port";
        String portString="{\"port\":"+port+"}";
        Message message=new Message(portString.getBytes(),topic, MessageUtil.messageId(),0);
        publish(topic,message);
    }

    @Override
    public void onDataRead(Connection connection, Stream stream, StreamData streamData) {
        Message m = this.convertStreamData2Message(streamData);
        log.info("receive msg, messageId:{}, data size: {}", m.getMessageId(), m.getPayload().length);
        CompletableFuture<MessageCallback.Action> cf = CompletableFuture.supplyAsync(() -> {
            try {
                Optional<MessageCallback> op = this.subscribeInfos.stream().filter((info) -> info.isMatch(m.getTopic())).findFirst().map(SubscribeInfo::getCallback);
                MessageCallback callback = op.orElse(this.messageCallback);
                if (callback != null) {
                    return this.messageCallback.consume(m);
                } else {
                    log.warn("no message callback for " + m.getTopic());
                    return MessageCallback.Action.CommitFailure;
                }
            } catch (Throwable var5) {
                log.error("message consume error, messageId:{}, {}", m.getMessageId(), var5);
                return MessageCallback.Action.CommitFailure;
            }
        });
        if (m.needAck()) {
            cf.whenComplete((action, throwable) -> {
                if (throwable != null) {
                    log.error("consume message {}, occurs error: ", m, throwable.getMessage());
                }
                log.info("consume message: {} , result: {}", m, action == null ? "null" : action.name());
                if (action == MessageCallback.Action.CommitSuccess) {
                    this.ack(m);
                }
            });
        }
    }

    @Override
    public void ack(Message message) {
        HttpHeader headers = new HttpHeader();
        int messageId =message.getMessageId();
        headers.put("x-message-id", messageId);
        headers.path("/message/ack");
        this.client().sendRequest(connection(), headers,null).whenComplete((r, t) -> {
            if (t != null) {
                log.error("ack failed, messageId {}, error {}", messageId, t.getMessage());
            } else {
                if (r.isOk()) {
                    log.debug("ack success, messageId {}", messageId);
                } else {
                    log.error((new ClientException("ack message: " + messageId + " failed", r)).getMessage());
                }
            }
        });
    }

}
